package zpump

import (
	"context"
	"fmt"
	"sync"
	"time"

	"gitee.com/youkelike/ziam/zanalytics"
	"gitee.com/youkelike/ziam/zpump/config"
	"gitee.com/youkelike/ziam/zpump/options"
	"gitee.com/youkelike/ziam/zpump/pumps"
	"gitee.com/youkelike/ziam/zpump/storage"
	"gitee.com/youkelike/ziam/zpump/storage/redis"
	"gitee.com/youkelike/zlog"
	"github.com/go-redsync/redsync/v4"
	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
	goredislib "github.com/redis/go-redis/v9"
	"gopkg.in/vmihailenco/msgpack.v2"
)

var pmps []pumps.Pump

type pumpServer struct {
	secInterval int
	omitDetails bool
	// 分布式锁
	mutex *redsync.Mutex
	// redis 做的数据仓库
	analyticsStore storage.AnalyticsStorage
	// 不同 pump 的配置
	pumps map[string]options.PumpConfig
}

type preparedPumpServer struct {
	*pumpServer
}

func createPumpServer(cfg *config.Config) (*pumpServer, error) {
	// 利用 redis 创建一个分布式锁
	client := goredislib.NewClient(&goredislib.Options{
		Addr:     fmt.Sprintf("%s:%d", cfg.RedisOptions.Host, cfg.RedisOptions.Port),
		Username: cfg.RedisOptions.Username,
		Password: cfg.RedisOptions.Password,
	})
	rs := redsync.New(goredis.NewPool(client))
	mutex := rs.NewMutex("ziam-pump", redsync.WithExpiry(10*time.Minute))

	// 封装了 redis 客户端的对象
	analyticsStore := &redis.RedisClusterStorageManager{}
	// 把选项参数传给 redis，用于后续的创建客户端连接
	if err := analyticsStore.Init(cfg.RedisOptions); err != nil {
		return nil, err
	}

	server := &pumpServer{
		secInterval:    cfg.PurgeDelay,
		omitDetails:    cfg.OmitDetailedRecording,
		mutex:          mutex,
		analyticsStore: analyticsStore,
		pumps:          cfg.Pumps,
	}

	return server, nil
}

func (s *pumpServer) PrepareRun() preparedPumpServer {
	// 过滤合法的 pump，然后初始化每个 pump
	s.initialize()

	return preparedPumpServer{s}
}

func (s *pumpServer) initialize() {
	// 把配置文件中配置的插件进行初始化
	pmps = make([]pumps.Pump, len(s.pumps))
	i := 0
	for key, pmpConf := range s.pumps {
		name := pmpConf.Type
		if name == "" {
			name = key
		}
		pmp, err := pumps.GetPumpByName(name)
		if err != nil {
			zlog.Errorw("pump load error (skipping): %s", err.Error())
		} else {
			// 重新构造一个 pmp，避免污染 s.pumps
			pmpIns := pmp.New()
			if err := pmpIns.Init(pmpConf.Meta); err != nil {
				zlog.Errorw("pump init error (skipping): %s", err.Error())
				continue
			}

			pmpIns.SetFilters(pmpConf.Filters)
			pmpIns.SetTimeout(pmpConf.Timeout)
			pmpIns.SetOmitDetailedRecording(pmpConf.OmitDetailedRecording)
			pmps[i] = pmpIns
		}
		i++
	}
}

// 启动定时任务
func (s preparedPumpServer) Run(stopCh <-chan struct{}) error {
	ticker := time.NewTicker(time.Duration(s.secInterval) * time.Second)
	defer ticker.Stop()

	zlog.Infow("start run loop to clean data from redis")
	for {
		select {
		case <-ticker.C:
			s.pump()
		case <-stopCh:
			zlog.Infow("stop purge loop")
			return nil
		}
	}
}

func (s *preparedPumpServer) pump() {
	// 拿到分布式锁
	if err := s.mutex.Lock(); err != nil {
		zlog.Infow("there is already an pump instance running.")
		return
	}

	defer func() {
		if _, err := s.mutex.Unlock(); err != nil {
			zlog.Errorw("could not release pump lock. err: %v", err)
		}
	}()

	ctx, _ := context.WithTimeout(context.Background(), time.Second*5)

	// 数据抓取
	// 从 redis 中获取数据，一次性抓取所有数据
	analyticsValues := s.analyticsStore.GetAndDeleteSet(ctx, zanalytics.AnalyticsKeyName)
	if len(analyticsValues) == 0 {
		return
	}

	// 数据预清洗
	// 数据解析到固定格式后，进行全局字段筛选，后面还有针对单个 pump 的字段筛选
	items := make([]interface{}, len(analyticsValues))
	for i, v := range analyticsValues {
		// redis 中存放的数据就是这个格式的
		decoded := zanalytics.AnalyticsRecord{}
		// msgpack 比 json 更小更快
		err := msgpack.Unmarshal([]byte(v.(string)), &decoded)
		if err != nil {
			zlog.Errorw("unmarshal analytics data failed: %s", err.Error())
		} else {
			if s.omitDetails {
				decoded.Policies = ""
				decoded.Deciders = ""
			}
			items[i] = interface{}(decoded)
		}
	}

	// 数据并发上报
	writeToPumps(items, s.secInterval)
}

func writeToPumps(items []interface{}, purgeDelay int) {
	if pmps == nil {
		zlog.Warnw("no pumps defined!")
		return
	}

	// 多个 pump 并发上报
	var wg sync.WaitGroup
	wg.Add(len(pmps))
	for _, p := range pmps {
		go execPumpWrite(&wg, p, &items, purgeDelay)
	}
	wg.Wait()
}

func execPumpWrite(wg *sync.WaitGroup, pmp pumps.Pump, items *[]interface{}, purgeDelay int) {
	timer := time.AfterFunc(time.Duration(purgeDelay)*time.Second, func() {
		// 能执行到这里，说明后面的上报任务还没结束
		if pmp.GetTimeout() == 0 {
			zlog.Warnw(
				"Pump %s is taking more time than the value configured of purge_delay. You should try to set a timeout for this pump.",
				pmp.GetName(),
			)
		} else if pmp.GetTimeout() > purgeDelay {
			zlog.Warnw("Pump %s is taking more time purge_delay. You should try lowering the timeout configured for this pump.", pmp.GetName())
		}
	})
	defer timer.Stop()
	defer wg.Done()

	zlog.Debugw("writing to: %s", pmp.GetName())

	ch := make(chan error, 1)
	var ctx context.Context
	var cancel context.CancelFunc
	if tm := pmp.GetTimeout(); tm > 0 {
		ctx, cancel = context.WithTimeout(context.Background(), time.Second*time.Duration(tm))
	} else {
		ctx, cancel = context.WithCancel(context.Background())
	}
	defer cancel()

	// 在 goroutine 中执行上报
	go func(ch chan error, ctx context.Context, pmp pumps.Pump, items *[]interface{}) {
		filteredItems := filterData(pmp, *items)
		ch <- pmp.WriteData(ctx, filteredItems)
	}(ch, ctx, pmp, items)

	// 获取上报结果和超时控制
	select {
	case err := <-ch:
		if err != nil {
			zlog.Warnw("error writing to: %s - Error: %s", pmp.GetName(), err.Error())
		}
	case <-ctx.Done():
		switch ctx.Err() {
		case context.Canceled:
			zlog.Warnw("writing to %s have got canceled.", pmp.GetName())
		case context.DeadlineExceeded:
			zlog.Warnw("timeout writing to: %s", pmp.GetName())
		}
	}
}

// 每个 pump 内部的数据清洗
func filterData(pmp pumps.Pump, items []interface{}) []interface{} {
	filters := pmp.GetFilters()
	if !filters.HasFilter() && !pmp.GetOmitDetailedRecording() {
		return items
	}

	filteredItems := items[:]
	newLen := 0

	for _, item := range items {
		decoded, _ := item.(zanalytics.AnalyticsRecord)
		// 过滤字段
		if pmp.GetOmitDetailedRecording() {
			decoded.Policies = ""
			decoded.Deciders = ""
		}

		// 白名单、黑名单
		if filters.ShouldFilter(decoded) {
			continue
		}

		filteredItems[newLen] = decoded
		newLen++
	}

	return filteredItems
}
